#include <assert.h>
#include <errno.h>
#include <netinet/in.h>
#include <search.h>
#include <signal.h>
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <unistd.h>
#include <sys/eventfd.h>

#include "shm_ipc.h"
#ifdef _XOPEN_SOURCE_EXTENDED
#include <arpa/inet.h>
#endif
#include <pthread.h>
#include <sys/socket.h>

#include <map>

#include "conn_node_client.h"
#include "deamon.h"
#include "game_event.h"
#include "game_thread.h"
#include "kafka_thread.h"
#include "oper_config.h"
#include "time_helper.h"
#include "msg_queue.h"
#include "librdkafka/rdkafka.h"
//#include "../proto/db.pb-c.h"

#define TIMEOUT_MISEC 10
char *g_game_topic;
int g_game_partid;

static void stop(int sig)
{
	g_thread_run = 0;
	global_el->stop = 1;
	wakeup_send_thread(1);
}

static int init_conn_client_map()
{
	return (0);
}

static int conn_kafka_recv_callback(struct rd_kafka_message_s *rkm)
{
// int send_one_buffer(conn_node_base *node, char *buffer, uint32_t len)
	std::map<int, conn_node_client *>::iterator ite;
	for (ite = conn_node_client::map_fd_nodes.begin(); ite != conn_node_client::map_fd_nodes.end(); ++ite)
	{
		send_one_buffer(ite->second, (char *)rkm->payload, rkm->len);
	}
	return (0);
}

static int cb_connsrv_timer(struct aeEventLoop *eventLoop, long long id, void *clientData)
{
	//	LOG_DEBUG("%s: id = %lld", __FUNCTION__, id);
	//	printf("%s id = %lld\n", __FUNCTION__, id);
	return (TIMEOUT_MISEC);
}

// int main_eventfd;
// static void cb_eventfd_func(aeEventLoop *el, int fd, void *privdata, int mask)
// {
// 	uint64_t id;
// 	assert(fd == main_eventfd);
// 	int ret = read(fd, &id, sizeof(uint64_t));
// 	if (ret != sizeof(uint64_t))
// 	{
// 		LOG_ERR("%s: read failed", __FUNCTION__);
// 		return;
// 	}
// 	struct message_queue *q = normal_queue_getmq(&main_queue);
// 	while (q) {
// 		rd_kafka_message_t *rkm = (rd_kafka_message_t *)q->data;

// 		printf(" Key: %.*s\n",
// 		       (int)rkm->len, (const char *)rkm->payload);
// 		rd_kafka_message_destroy(rkm);

// 		q = q->next;
// 	}
// }


int main(int argc, char **argv)
{
	int   ret  = 0;
	FILE *file = NULL;
	char *line;
	int   port;
//	int   game_partid;
//	char *game_topic;
	char *game_brokers;
	int   conn_partid;
	char *conn_topic;
	char *conn_brokers;
	pthread_t kafka_recv_pid;
	pthread_t kafka_send_pid;	

	signal(SIGTERM, SIG_IGN);

	ret = log4c_init();
	if (ret != 0)
	{
		printf("log4c_init failed[%d]\n", ret);
		return (ret);
	}

	init_mycat();
	if (!mycat)
	{
		printf("log4c_category_get(\"six13log.log.app.application1\"); failed\n");
		return (0);
	}
	for (int i = 1; i < argc; ++i)
	{
		if (strcmp(argv[i], "-d") == 0)
		{
			change_to_deamon();
			//			break;
		}
		else if (strcmp(argv[i], "-t") == 0)
		{  /// test for mem check
			open_err_log_file();
		}
	}

	uint64_t pid = write_pid_file();
	LOG_INFO("%s %d: conn_srv run %lu", __FUNCTION__, __LINE__, pid);
	if (init_conn_client_map() != 0)
	{
		LOG_ERR("init client map failed");
		goto done;
	}

	ret = game_event_init();
	if (ret != 0)
		goto done;

	file = fopen("../server_info.ini", "r");
	if (!file)
	{
		LOG_ERR("open server_info.ini failed[%d]", errno);
		ret = -1;
		goto done;
	}

	line = get_first_key(file, (char *)"game_topic");
	g_game_topic = get_value(line);
	line = get_first_key(file, (char *)"game_partid");
	g_game_partid = atoi(get_value(line));
	line = get_first_key(file, (char *)"game_broker");
	game_brokers = get_value(line);
	line = get_first_key(file, (char *)"conn_topic");
	conn_topic = get_value(line);
	line = get_first_key(file, (char *)"conn_partid");
	conn_partid = atoi(get_value(line));
	line = get_first_key(file, (char *)"conn_broker");
	conn_brokers = get_value(line);

	line = get_first_key(file, (char *)"conn_client_port");
	port = atoi(get_value(line));
	if (port <= 0)
	{
		LOG_ERR("config file wrong, no conn_srv_client_port");
		ret = -1;
		goto done;
	}

	ret = game_add_listen_event(NULL, port, conn_node_client::get_conn_node, "client");
	if (ret < 0)
		goto done;
	conn_node_client::listen_fd = ret;
	if (ret < 0)
		goto done;

	if (SIG_ERR == signal(SIGPIPE, SIG_IGN))
	{
		LOG_ERR("set sigpipe ign failed");
		return (0);
	}
	signal(SIGINT, stop);
	signal(SIGUSR2, stop);	

	// aeSetAfterSleepProc(global_el, cb_recv_shm_ipcs);
	aeCreateTimeEvent(global_el, TIMEOUT_MISEC, cb_connsrv_timer, NULL, NULL);

	kafka_recv_pid = create_kafka_recv_thread(conn_brokers, (char *)"1", conn_partid, conn_topic, conn_kafka_recv_callback, NULL);
	kafka_send_pid = create_kafka_send_thread(game_brokers, NULL);	

	aeMain(global_el);
	
	pthread_join(kafka_recv_pid, NULL);
	pthread_join(kafka_send_pid, NULL);	

	aeDeleteEventLoop(global_el);

	LOG_INFO("srv loop stoped[%d]", ret);

done:
	LOG_INFO("conn_srv stoped[%d]", ret);
	if (file)
		fclose(file);
	return (ret);
}
